//
//  hdfs_client.cpp
//  DataPlatform
//
//  Created by Liu Dafan on 12-2-3.
//  Copyright (c) 2012年 __MyCompanyName__. All rights reserved.
//

#include <iostream>
#include "error_code.h"
#include "hdfs_client.h"

using namespace std;

HdfsOp::HdfsOp(char *sv_ip, unsigned short sv_port)
{
	*_sv_ip = '\0';
	strncpy(_sv_ip, sv_ip, INET_ADDRSTRLEN);
	_sv_port = sv_port;
	_hclient = NULL;
    _isInit = false;
}

HdfsOp::~HdfsOp()
{
    if (_hclient != NULL) {
        delete _hclient;
    }
}

int HdfsOp::init()
{
	if (_hclient) {
		DERROR("hbase open error: already open\n");
		return ERROR_CODE_OK;
	}
    _socket.reset(new TSocket(_sv_ip, _sv_port));
    _transport.reset(new TBufferedTransport(_socket));
    _protocol.reset(new TBinaryProtocol(_transport));
    
    _hclient = new ThriftHadoopFileSystemClient(_protocol);
	try {
		_transport->open();
	} catch (TException &e) {
		DERROR("hdfs connect error: errrno:%d:%s\n", _errnum, e.what());
		return ERROR_CODE_INTERNAL;
	}

    _isInit = true;
    
    return ERROR_CODE_OK;
}

void HdfsOp::uninit()
{
    if (_isInit)
    {
        _transport->close();
        delete _hclient;
    }
    
    _isInit = false;
}

HdfsHandle HdfsOp::open(const string& fileName, FILE_MODE mode)
{
    if (!_isInit || _hclient == NULL)
    {
        DERROR("Doesn't initialize\n");
        return NULL;
    }
    
    ThriftHandle h;
    Pathname path;
    path.pathname = fileName;

    try {
        switch (mode) {
            case FILE_MODE_CREATE:
                _hclient->create(h, path);
                break;
            case FILE_MODE_READ:
                _hclient->open(h, path);
                break;
            case FILE_MODE_APPEND:
                _hclient->append(h, path);
                break;
        }
    } catch (TException &tx) {
        DERROR("open file Failed due to %s, filename: %s, mode: %d",
               tx.what(), fileName.c_str(), mode);
        return NULL;
    }
    
    return new HdfsFile(h);
}

void HdfsOp::close(HdfsHandle handle)
{
    if (handle == NULL)
    {
        return;
    }

    try {
        _hclient->close(handle->_handle);
    } catch (TException &tx) {
        DERROR("Write bytes Failed due to %s", tx.what());
        return;
    }
    
    delete handle;
}

int HdfsOp::writebytes(HdfsHandle handle, const int8_t* data, size_t size)
{
    if (!_isInit || _hclient == NULL)
    {
        DERROR("Doesn't initialize\n");
        return ERROR_CODE_INTERNAL;
    }
    
    if (handle == NULL)
    {
        DERROR("Invalid Handle");
        return ERROR_CODE_ILLEGAL_ARGUMENT;
    }
    
    vector<int8_t> vec_data;
    
    for (size_t idx = 0; idx < size; idx++)
    {
        vec_data.push_back(data[idx]);
    }
    
    return writebytes(handle, vec_data);
}

int HdfsOp::writebytes(HdfsHandle handle, const vector<int8_t>& vec_data)
{
    if (!_isInit || _hclient == NULL)
    {
        DERROR("Doesn't initialize\n");
        return ERROR_CODE_INTERNAL;
    }
    
    if (handle == NULL)
    {
        DERROR("Invalid Handle");
        return ERROR_CODE_ILLEGAL_ARGUMENT;
    }
    
    try {
        _hclient->writebytes(handle->_handle, vec_data);
    } catch (TException &tx) {
        DERROR("Write bytes Failed due to %s", tx.what());
        return ERROR_CODE_INTERNAL;
    }
    
    return ERROR_CODE_OK;    
}

int HdfsOp::readbytes(HdfsHandle handle, int offset, long length, int8_t*& data, size_t& size)
{
    if (!_isInit || _hclient == NULL)
    {
        DERROR("Doesn't initialize\n");
        return ERROR_CODE_INTERNAL;
    }
    
    if (handle == NULL)
    {
        DERROR("Invalid Handle");
        return ERROR_CODE_ILLEGAL_ARGUMENT;
    }
    
    if (data != NULL)
    {
        DERROR("The input data is not null");
        return ERROR_CODE_ILLEGAL_ARGUMENT;        
    }
    
    vector<int8_t> vec_data;
    try {
        _hclient->readbytes(vec_data, handle->_handle, offset, length);
    } catch (TException &tx) {
        DERROR("Read bytes Failed due to %s", tx.what());
        return ERROR_CODE_INTERNAL;
    }

    data = new int8_t[vec_data.size()];
    for (size_t idx = 0; idx < vec_data.size(); idx ++)
    {
        data[idx] = vec_data[idx];
    }
    size = vec_data.size();
    
    return ERROR_CODE_OK;
}

int HdfsOp::readbytes(HdfsHandle handle, int offset, long length, vector<int8_t>& vec_data)
{
    if (!_isInit || _hclient == NULL)
    {
        DERROR("Doesn't initialize\n");
        return ERROR_CODE_INTERNAL;
    }
    
    if (handle == NULL)
    {
        DERROR("Invalid Handle");
        return ERROR_CODE_ILLEGAL_ARGUMENT;
    }
    
    try {
        _hclient->readbytes(vec_data, handle->_handle, offset, length);
    } catch (TException &tx) {
        DERROR("Read bytes Failed due to %s", tx.what());
        return ERROR_CODE_INTERNAL;
    }
    
    return ERROR_CODE_OK;
}

int HdfsOp::getlength(const string& fileName, uint64_t& length)
{
    if (!_isInit || _hclient == NULL)
    {
        DERROR("Doesn't initialize\n");
        return ERROR_CODE_INTERNAL;
    }
    
    FileStatus status;
    Pathname path;
    path.pathname = fileName;
    
    try {
        _hclient->stat(status, path);
    } catch (TException &tx) {
        DERROR("Read bytes Failed due to %s", tx.what());
        return ERROR_CODE_INTERNAL;
    }
    
    length = status.length;
    
    return ERROR_CODE_OK;

}

int HdfsOp::listfiles(const string& pathName, vector<FileStatus>& files)
{
    files.resize(0);
    
    Pathname path;
    path.pathname = pathName;
    
    try {
        _hclient->listStatus(files, path);
    } catch (TException &tx) {
        DERROR("Read bytes Failed due to %s", tx.what());
        return ERROR_CODE_INTERNAL;
    }
    
    return ERROR_CODE_OK;
}

bool HdfsOp::isExist(const string& filename)
{
    Pathname path;
    path.pathname = filename;
 
    try {
        return _hclient->exists(path);
    } catch (TException &tx) {
        DERROR("Read bytes Failed due to %s", tx.what());
        //TODO: 
        return false;
    }
}
